package com.adu.music.spider;

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class SpiderRunnerPool {
    /**
     * 任务队列容量
     */
    private int taskCapacity;
    /**
     * 线程数量
     */
    private int threadCount;

    private Thread[] thrds;
    /**
     * 任务队列
     */
    private LinkedList<Runnable> taskQueue = new LinkedList<>();

    private Lock lock = new ReentrantLock();

    private Condition fullCond = lock.newCondition();
    private Condition emptyCond = lock.newCondition();
    /**
     * 是否停止添加任务
     */
    private volatile boolean stoped = false;

    public SpiderRunnerPool(int threadCount, int taskCapacity) {
        this.threadCount = threadCount;
        this.taskCapacity = taskCapacity;
        thrds = new Thread[this.threadCount];
        this.run();
    }

    public SpiderRunnerPool(int threadCount) {
        this(threadCount, threadCount * 10);
    }

    public void run() {
        for (int i = 0; i < this.threadCount; i++) {
            thrds[i] = new Thread(() -> {
                Runnable task;
                while (true) {
                    lock.lock();
                    try {
                        while (!stoped && taskQueue.isEmpty()) {
                            try {
                                emptyCond.await();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        if (taskQueue.size() >= taskCapacity) {
                            fullCond.signalAll();
                        }
                        task = taskQueue.poll();
                    } finally {
                        lock.unlock();
                    }
                    if (task != null) {
                        task.run();
                    }
                    if (stoped && taskQueue.isEmpty()) {
                        break;
                    }
                }
            }, "Thread-" + i);
            thrds[i].start();
        }
    }

    public void execute(Runnable task) {
        lock.lock();
        try {
            if (!stoped) {
                while (taskQueue.size() >= this.taskCapacity) {
                    try {
                        fullCond.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //if(taskQueue.isEmpty()){
                emptyCond.signalAll();
                //}
                taskQueue.add(task);
            } else {
                throw new RuntimeException("线程池已经关闭，无法再添加新的任务");
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 表示无法再添加新的任务到队列中，线程池会将剩余任务执行完毕
     */
    public void shutdown() {
        lock.lock();
        try {
            stoped = true;
            fullCond.signalAll();
            emptyCond.signalAll();
        } finally {
            lock.unlock();
        }
    }
}
